package com.teradata.multithread.dsrptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ThreadFactory;

/**
 * @author: lyndon
 * @version: 1.0
 * @date: 2019/2/18
 */


public class DisruptorManager {

    private Disruptor<MessageEvent> disruptor;
    private RingBuffer<MessageEvent> ringBuffer;

    public DisruptorManager(SftpGetFileHandler sftpGetFileHandler, InsertDbHandler insertDBHandler) {
        System.out.println("初始化Disruptor开始。。。");
        ThreadFactory threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r,"MESSAGE-PUBLISH-THREAD");
            }
        };
        disruptor = new Disruptor<>(new MessageEventFactory(), 8 * 1024, threadFactory, ProducerType.MULTI, new BlockingWaitStrategy());
        ringBuffer = disruptor.getRingBuffer();
        disruptor.handleEventsWith(sftpGetFileHandler)
                .then(insertDBHandler)
        ;
        disruptor.start();
        System.out.println("初始化Disruptor结束。。。");
    }

    public void destroy() {
        if(disruptor != null) {
            disruptor.shutdown();
        }
    }

    public void work(Message message) {
        if(null == ringBuffer) {
            return;
        }
        long next = ringBuffer.next();
        try {
            MessageEvent messageEvent = ringBuffer.get(next);
            messageEvent.setMessage(message);
        } finally {
            ringBuffer.publish(next);
        }
    }
}
